<!doctype html>
<html lang="en">
<head>
    <meta charset="utf-8">
    <meta name="viewport" content="width=device-width, initial-scale=1.0">
    <meta name="description" content="coinfaces">

        <title>RxJS学习笔记</title>

    <link rel="stylesheet" href="../../theme/css/style.css">
    <link rel="stylesheet" href="../../theme/css/pygments.css">



</head>

<body>
    <div class="content">
<h1>RxJS学习笔记</h1>    <p>
        under
            <a href="../../tags/rxjs/">rxjs</a>
    </p>
    <p>
        in <a href="../../categories/tech/">tech</a>
    </p>
    <p>Published: 2019-12-02</p>


    <p>该笔记基于RxJS 6</p>
<h4>基础概念</h4>
<p><a href="https://www.learnrxjs.io/concepts/rxjs-primer.html">RxJS Primer</a>介绍了RxJS中常见概念和基本用法。</p>
<p><a href="https://medium.com/@AnkurRatra/demystifying-rxjs-observable-467c52309ac">Demystifying RxJS Observable</a>介绍了RxJS Observable的概念模型和简单实现。简单地说，RxJS Observable本质上是一个函数，接受Observer作为其参数，当其被调用时，它调用Observer的<code>next</code>、<code>error</code>和<code>complete</code>方法处理数据，最后用类封装出API。这个模型有助于理解Observable为什么默认是“冷的”、“懒的”。</p>
<div class="highlight"><pre><span></span><code><span class="c1">// 举例：根据此思想实现的`of`函数</span>
<span class="kr">interface</span> <span class="nx">IObserver</span> <span class="p">{</span>
  <span class="nx">next</span><span class="o">:</span> <span class="p">(</span><span class="nx">data</span>: <span class="kt">any</span><span class="p">)</span> <span class="o">=&gt;</span> <span class="k">void</span>
  <span class="nx">error</span><span class="o">:</span> <span class="p">(</span><span class="nx">error</span>: <span class="kt">Error</span><span class="p">)</span> <span class="o">=&gt;</span> <span class="k">void</span>
  <span class="nx">complete</span><span class="o">:</span> <span class="p">()</span> <span class="o">=&gt;</span> <span class="k">void</span>
<span class="p">}</span>

<span class="nx">type</span> <span class="nx">IPartialObserver</span> <span class="o">=</span> <span class="nx">Partial</span><span class="o">&lt;</span><span class="nx">IObserver</span><span class="o">&gt;</span><span class="p">;</span>

<span class="nx">type</span> <span class="nx">ICleanupLogic</span> <span class="o">=</span> <span class="p">()</span> <span class="o">=&gt;</span> <span class="k">void</span>

<span class="nx">type</span> <span class="nx">IObservable</span> <span class="o">=</span> <span class="p">(</span><span class="nx">observer</span>: <span class="kt">IPartialObserver</span><span class="p">)</span> <span class="o">=&gt;</span> <span class="nx">ICleanupLogic</span>

<span class="kr">const</span> <span class="nx">patch</span> <span class="o">=</span> <span class="p">(</span><span class="nx">partialObserver</span>: <span class="kt">IPartialObserver</span><span class="p">)</span><span class="o">:</span> <span class="nx">IObserver</span> <span class="o">=&gt;</span> <span class="p">{</span>
  <span class="k">return</span> <span class="p">{</span>
    <span class="nx">next</span>: <span class="kt">partialObserver.next</span> <span class="o">||</span> <span class="p">(()</span> <span class="o">=&gt;</span> <span class="p">{}),</span>
    <span class="nx">error</span>: <span class="kt">partialObserver.error</span> <span class="o">||</span> <span class="p">(()</span> <span class="o">=&gt;</span> <span class="p">{}),</span>
    <span class="nx">complete</span>: <span class="kt">partialObserver.complete</span> <span class="o">||</span> <span class="p">(()</span> <span class="o">=&gt;</span> <span class="p">{})</span>
  <span class="p">};</span>
<span class="p">};</span>

<span class="kr">const</span> <span class="k">of</span> <span class="o">=</span> <span class="p">(...</span><span class="nx">args</span>: <span class="kt">any</span><span class="p">[])</span><span class="o">:</span> <span class="nx">IObservable</span> <span class="o">=&gt;</span> <span class="p">{</span>
  <span class="k">return</span> <span class="p">(</span><span class="nx">observer</span>: <span class="kt">IPartialObserver</span><span class="p">)</span> <span class="o">=&gt;</span> <span class="p">{</span>
    <span class="nx">args</span><span class="p">.</span><span class="nx">forEach</span><span class="p">((</span><span class="nx">data</span><span class="p">)</span> <span class="o">=&gt;</span> <span class="nx">patch</span><span class="p">(</span><span class="nx">observer</span><span class="p">).</span><span class="nx">next</span><span class="p">(</span><span class="nx">data</span><span class="p">));</span>
    <span class="k">return</span> <span class="p">()</span> <span class="o">=&gt;</span> <span class="p">{</span>
      <span class="nx">console</span><span class="p">.</span><span class="nx">log</span><span class="p">(</span><span class="s1">&#39;Clean up&#39;</span><span class="p">);</span>
    <span class="p">};</span>
  <span class="p">}</span>
<span class="p">};</span>

<span class="c1">// Create source</span>
<span class="kr">const</span> <span class="nx">source</span> <span class="o">=</span> <span class="k">of</span><span class="p">(</span><span class="mi">1</span><span class="p">,</span> <span class="mi">2</span><span class="p">,</span> <span class="mi">3</span><span class="p">,</span> <span class="mi">4</span><span class="p">);</span>

<span class="c1">// Subscribe</span>
<span class="kr">const</span> <span class="nx">unsub</span> <span class="o">=</span> <span class="nx">source</span><span class="p">({</span>
  <span class="nx">next</span><span class="o">:</span> <span class="p">(</span><span class="nx">x</span><span class="p">)</span> <span class="o">=&gt;</span> <span class="nx">console</span><span class="p">.</span><span class="nx">log</span><span class="p">(</span><span class="nx">x</span><span class="p">)</span>
<span class="p">});</span>

<span class="c1">// Unsubscribe</span>
<span class="nx">unsub</span><span class="p">();</span>

<span class="c1">// 输出：</span>
<span class="c1">//</span>
<span class="c1">// 1</span>
<span class="c1">// 2</span>
<span class="c1">// 3</span>
<span class="c1">// 4</span>
<span class="c1">// Clean up</span>
</code></pre></div>


<p>同样也可以创建一个“热的”Observable：</p>
<div class="highlight"><pre><span></span><code><span class="c1">// interval和ConnectableObserver</span>
<span class="kr">const</span> <span class="nx">interval</span> <span class="o">=</span> <span class="p">(</span><span class="nx">ms</span>: <span class="kt">number</span><span class="p">)</span> <span class="o">=&gt;</span> <span class="p">{</span>
  <span class="k">return</span> <span class="p">(</span><span class="nx">observer</span>: <span class="kt">IPartialObserver</span><span class="p">)</span> <span class="o">=&gt;</span> <span class="p">{</span>
    <span class="kd">let</span> <span class="nx">value</span> <span class="o">=</span> <span class="mi">0</span><span class="p">;</span>
    <span class="kd">let</span> <span class="nx">timeoutHandle</span>: <span class="kt">number</span><span class="p">;</span>
    <span class="kr">const</span> <span class="nx">emit</span> <span class="o">=</span> <span class="p">()</span> <span class="o">=&gt;</span> <span class="p">{</span>
      <span class="nx">timeoutHandle</span> <span class="o">=</span> <span class="nx">setTimeout</span><span class="p">(()</span> <span class="o">=&gt;</span> <span class="p">{</span>
        <span class="nx">patch</span><span class="p">(</span><span class="nx">observer</span><span class="p">).</span><span class="nx">next</span><span class="p">(</span><span class="nx">value</span><span class="p">);</span>
        <span class="nx">value</span> <span class="o">+=</span> <span class="mi">1</span><span class="p">;</span>
        <span class="nx">emit</span><span class="p">();</span>
      <span class="p">},</span> <span class="nx">ms</span><span class="p">);</span>
    <span class="p">};</span>
    <span class="nx">emit</span><span class="p">();</span>
    <span class="k">return</span> <span class="p">()</span> <span class="o">=&gt;</span> <span class="p">{</span>
      <span class="k">if</span> <span class="p">(</span><span class="nx">timeoutHandle</span> <span class="o">!=</span> <span class="kc">null</span><span class="p">)</span> <span class="p">{</span>
        <span class="nx">clearTimeout</span><span class="p">(</span><span class="nx">timeoutHandle</span><span class="p">);</span>
      <span class="p">}</span>
    <span class="p">};</span>
  <span class="p">};</span>
<span class="p">};</span>

<span class="nx">type</span> <span class="nx">IConnectable</span> <span class="o">=</span> <span class="nx">IObservable</span> <span class="o">&amp;</span> <span class="p">((</span><span class="nx">action</span><span class="o">:</span> <span class="s1">&#39;connect&#39;</span><span class="p">)</span> <span class="o">=&gt;</span> <span class="nx">ICleanupLogic</span><span class="p">);</span>

<span class="kr">const</span> <span class="nx">makeConnectable</span> <span class="o">=</span> <span class="p">(</span><span class="nx">source</span>: <span class="kt">IObservable</span><span class="p">)</span> <span class="o">=&gt;</span> <span class="p">{</span>
  <span class="kr">const</span> <span class="nx">partialObservers</span>: <span class="kt">IPartialObserver</span><span class="p">[]</span> <span class="o">=</span> <span class="p">[];</span>
  <span class="k">return</span> <span class="p">(</span><span class="nx">actionOrObserver</span><span class="o">:</span> <span class="s1">&#39;connect&#39;</span> <span class="o">|</span> <span class="nx">IPartialObserver</span><span class="p">)</span> <span class="o">=&gt;</span> <span class="p">{</span>
    <span class="k">if</span> <span class="p">(</span><span class="nx">actionOrObserver</span> <span class="o">===</span> <span class="s1">&#39;connect&#39;</span><span class="p">)</span> <span class="p">{</span>
      <span class="k">return</span> <span class="nx">source</span><span class="p">({</span>
        <span class="nx">next</span>: <span class="kt">value</span> <span class="o">=&gt;</span> <span class="nx">partialObservers</span><span class="p">.</span><span class="nx">forEach</span><span class="p">(</span><span class="nx">ob</span> <span class="o">=&gt;</span> <span class="nx">patch</span><span class="p">(</span><span class="nx">ob</span><span class="p">).</span><span class="nx">next</span><span class="p">(</span><span class="nx">value</span><span class="p">)),</span>
        <span class="nx">error</span>: <span class="kt">e</span> <span class="o">=&gt;</span> <span class="nx">partialObservers</span><span class="p">.</span><span class="nx">forEach</span><span class="p">(</span><span class="nx">ob</span> <span class="o">=&gt;</span> <span class="nx">patch</span><span class="p">(</span><span class="nx">ob</span><span class="p">).</span><span class="nx">error</span><span class="p">(</span><span class="nx">e</span><span class="p">)),</span>
        <span class="nx">complete</span><span class="o">:</span> <span class="p">()</span> <span class="o">=&gt;</span> <span class="nx">partialObservers</span><span class="p">.</span><span class="nx">forEach</span><span class="p">(</span><span class="nx">ob</span> <span class="o">=&gt;</span> <span class="nx">patch</span><span class="p">(</span><span class="nx">ob</span><span class="p">).</span><span class="nx">complete</span><span class="p">)</span>
      <span class="p">});</span>
    <span class="p">}</span> <span class="k">else</span> <span class="p">{</span>
      <span class="kr">const</span> <span class="nx">observer</span> <span class="o">=</span> <span class="nx">actionOrObserver</span><span class="p">;</span>
      <span class="k">if</span> <span class="p">(</span><span class="nx">partialObservers</span><span class="p">.</span><span class="nx">indexOf</span><span class="p">(</span><span class="nx">observer</span><span class="p">)</span> <span class="o">&lt;</span> <span class="mi">0</span><span class="p">)</span> <span class="p">{</span>
        <span class="nx">partialObservers</span><span class="p">.</span><span class="nx">push</span><span class="p">(</span><span class="nx">observer</span><span class="p">);</span>
      <span class="p">}</span>
      <span class="k">return</span> <span class="p">()</span> <span class="o">=&gt;</span> <span class="p">{</span>
        <span class="kr">const</span> <span class="nx">index</span> <span class="o">=</span> <span class="nx">partialObservers</span><span class="p">.</span><span class="nx">indexOf</span><span class="p">(</span><span class="nx">observer</span><span class="p">);</span>
        <span class="k">if</span> <span class="p">(</span><span class="nx">index</span> <span class="o">&gt;=</span> <span class="mi">0</span><span class="p">)</span> <span class="p">{</span>
          <span class="nx">partialObservers</span><span class="p">.</span><span class="nx">splice</span><span class="p">(</span><span class="nx">index</span><span class="p">,</span> <span class="mi">1</span><span class="p">);</span>
        <span class="p">}</span>
      <span class="p">}</span>
    <span class="p">}</span>
  <span class="p">};</span>
<span class="p">};</span>

<span class="kr">const</span> <span class="nx">connectable</span> <span class="o">=</span> <span class="nx">makeConnectable</span><span class="p">(</span><span class="nx">interval</span><span class="p">(</span><span class="mi">1000</span><span class="p">));</span>
<span class="c1">// Sub1准备处理数据，此时还没有数据产生，否则Sub2会错过前2个数据</span>
<span class="kr">const</span> <span class="nx">unsub1</span> <span class="o">=</span> <span class="nx">connectable</span><span class="p">({</span>
  <span class="nx">next</span>: <span class="kt">value</span> <span class="o">=&gt;</span> <span class="nx">console</span><span class="p">.</span><span class="nx">log</span><span class="p">(</span><span class="sb">`Sub1: </span><span class="si">${</span><span class="nx">value</span><span class="si">}</span><span class="sb">`</span><span class="p">)</span>
<span class="p">});</span>
<span class="nx">setTimeout</span><span class="p">(()</span> <span class="o">=&gt;</span> <span class="p">{</span>
  <span class="c1">// 开始产生数据</span>
  <span class="kr">const</span> <span class="nx">disconnect</span> <span class="o">=</span> <span class="nx">connectable</span><span class="p">(</span><span class="s1">&#39;connect&#39;</span><span class="p">);</span>
  <span class="nx">setTimeout</span><span class="p">(()</span> <span class="o">=&gt;</span> <span class="p">{</span>
    <span class="c1">// Sub2会错过第1个数据</span>
    <span class="kr">const</span> <span class="nx">unsub2</span> <span class="o">=</span> <span class="nx">connectable</span><span class="p">({</span>
      <span class="nx">next</span>: <span class="kt">value</span> <span class="o">=&gt;</span> <span class="nx">console</span><span class="p">.</span><span class="nx">log</span><span class="p">(</span><span class="sb">`Sub2: </span><span class="si">${</span><span class="nx">value</span><span class="si">}</span><span class="sb">`</span><span class="p">)</span>
    <span class="p">});</span>
    <span class="nx">setTimeout</span><span class="p">(()</span> <span class="o">=&gt;</span> <span class="p">{</span>
      <span class="c1">// Sub2停止处理数据，Sub1会继续处理后续数据</span>
      <span class="nx">unsub2</span><span class="p">();</span>
      <span class="nx">setTimeout</span><span class="p">(()</span> <span class="o">=&gt;</span> <span class="p">{</span>
        <span class="c1">// 停止产生数据，Sub1将不会收到任何数据</span>
        <span class="nx">disconnect</span><span class="p">();</span>
        <span class="nx">setTimeout</span><span class="p">(()</span> <span class="o">=&gt;</span> <span class="p">{</span>
          <span class="c1">// Sub1停止处理数据</span>
          <span class="nx">unsub1</span><span class="p">();</span>
        <span class="p">},</span> <span class="mi">1100</span><span class="p">);</span>
      <span class="p">},</span> <span class="mi">1100</span><span class="p">);</span>
    <span class="p">},</span> <span class="mi">1100</span><span class="p">);</span>
  <span class="p">},</span> <span class="mi">1100</span><span class="p">);</span>
<span class="p">},</span> <span class="mi">1100</span><span class="p">);</span>

<span class="c1">// 输出：</span>
<span class="c1">//</span>
<span class="c1">// Sub1: 0</span>
<span class="c1">// Sub1: 1</span>
<span class="c1">// Sub2: 1</span>
<span class="c1">// Sub1: 2</span>
</code></pre></div>


<h4>Observable对象的“冷热”</h4>
<p>Observable默认是“冷的”，这个“冷”有两种含义：</p>
<ol>
<li>只有其<code>.subscribe</code>方法被调用时才会产生数据</li>
<li>其对每个<code>.subscribe</code>调用产生独立的数据</li>
</ol>
<p><a href="https://blog.thoughtram.io/angular/2016/06/16/cold-vs-hot-observables.html">这篇文章</a>详细描述了“冷”和“热”Observable的区别和使用场景，这篇文章中的示例代码基于RxJS 5编写。</p>
<h4>publish运算符及其伙伴们</h4>
<p><code>publish</code>运算符返回一个<code>ConnectableObserver</code>对象，该对象在其<code>.connect</code>方法被调用后才开始产生数据，即使当前没有subscription，如果存在多个subscription，每个subscription都接收到来自同一数据流的相同数据。</p>
<div class="highlight"><pre><span></span><code><span class="kr">import</span> <span class="p">{</span> <span class="k">of</span><span class="p">,</span> <span class="nx">zip</span><span class="p">,</span> <span class="nx">interval</span><span class="p">,</span> <span class="nx">merge</span><span class="p">,</span> <span class="nx">ConnectableObservable</span> <span class="p">}</span> <span class="nx">from</span> <span class="s2">&quot;rxjs&quot;</span><span class="p">;</span>
<span class="kr">import</span> <span class="p">{</span> <span class="nx">map</span><span class="p">,</span> <span class="nx">publish</span><span class="p">,</span> <span class="nx">tap</span> <span class="p">}</span> <span class="nx">from</span> <span class="s2">&quot;rxjs/operators&quot;</span><span class="p">;</span>

<span class="nx">console</span><span class="p">.</span><span class="nx">clear</span><span class="p">();</span>

<span class="kr">const</span> <span class="nx">obs1</span> <span class="o">=</span> <span class="nx">interval</span><span class="p">(</span><span class="mi">1000</span><span class="p">);</span>
<span class="kr">const</span> <span class="nx">obs2</span> <span class="o">=</span> <span class="k">of</span><span class="p">(</span><span class="mi">1</span><span class="p">,</span> <span class="mi">2</span><span class="p">,</span> <span class="mi">3</span><span class="p">,</span> <span class="mi">4</span><span class="p">);</span>
<span class="kr">const</span> <span class="nx">obs$</span> <span class="o">=</span> <span class="nx">zip</span><span class="p">(</span><span class="nx">obs1</span><span class="p">,</span> <span class="nx">obs2</span><span class="p">).</span><span class="nx">pipe</span><span class="p">(</span>
  <span class="nx">map</span><span class="p">(</span><span class="nx">values</span> <span class="o">=&gt;</span> <span class="nx">values</span><span class="p">[</span><span class="mi">1</span><span class="p">])</span>
<span class="p">);</span>

<span class="kr">const</span> <span class="nx">published</span> <span class="o">=</span> <span class="nx">obs$</span>
  <span class="p">.</span><span class="nx">pipe</span><span class="p">(</span>
    <span class="nx">publish</span><span class="p">()</span>
  <span class="p">)</span>

<span class="nx">setTimeout</span><span class="p">(()</span> <span class="o">=&gt;</span> <span class="p">{</span>
  <span class="c1">// 该subscription没有接收到第一个数据</span>
  <span class="nx">published</span><span class="p">.</span><span class="nx">subscribe</span><span class="p">((</span><span class="nx">x</span><span class="p">)</span> <span class="o">=&gt;</span> <span class="nx">console</span><span class="p">.</span><span class="nx">log</span><span class="p">(</span><span class="sb">`Sub 1: </span><span class="si">${</span><span class="nx">x</span><span class="si">}</span><span class="sb">`</span><span class="p">));</span>
  <span class="nx">setTimeout</span><span class="p">(()</span> <span class="o">=&gt;</span> <span class="p">{</span>
    <span class="c1">// 该subscription没有接收到第1个和第2个数据</span>
    <span class="nx">published</span><span class="p">.</span><span class="nx">subscribe</span><span class="p">((</span><span class="nx">x</span><span class="p">)</span> <span class="o">=&gt;</span> <span class="nx">console</span><span class="p">.</span><span class="nx">log</span><span class="p">(</span><span class="sb">`Sub 2: </span><span class="si">${</span><span class="nx">x</span><span class="si">}</span><span class="sb">`</span><span class="p">));</span>
  <span class="p">},</span> <span class="mi">1100</span><span class="p">);</span>
<span class="p">},</span> <span class="mi">1100</span><span class="p">);</span>

<span class="c1">// connect之后马上开始产生数据</span>
<span class="nx">published</span><span class="p">.</span><span class="nx">connect</span><span class="p">();</span>

<span class="c1">// 输出</span>
<span class="c1">//</span>
<span class="c1">// Sub 1: 2</span>
<span class="c1">// Sub 1: 3</span>
<span class="c1">// Sub 2: 3</span>
<span class="c1">// Sub 1: 4</span>
<span class="c1">// Sub 2: 4</span>
</code></pre></div>


<p><code>publish</code>运算符还支持selector参数，这是一个函数，输入一个Observable对象，该对象的所有subscription接收同一个参数</p>
<div class="highlight"><pre><span></span><code><span class="kr">import</span> <span class="p">{</span> <span class="k">of</span><span class="p">,</span> <span class="nx">zip</span><span class="p">,</span> <span class="nx">interval</span><span class="p">,</span> <span class="nx">merge</span><span class="p">,</span> <span class="nx">ConnectableObservable</span> <span class="p">}</span> <span class="nx">from</span> <span class="s2">&quot;rxjs&quot;</span><span class="p">;</span>
<span class="kr">import</span> <span class="p">{</span> <span class="nx">map</span><span class="p">,</span> <span class="nx">publish</span><span class="p">,</span> <span class="nx">tap</span> <span class="p">}</span> <span class="nx">from</span> <span class="s2">&quot;rxjs/operators&quot;</span><span class="p">;</span>

<span class="nx">console</span><span class="p">.</span><span class="nx">clear</span><span class="p">();</span>

<span class="kr">const</span> <span class="nx">obs1</span> <span class="o">=</span> <span class="nx">interval</span><span class="p">(</span><span class="mi">1000</span><span class="p">);</span>
<span class="kr">const</span> <span class="nx">obs2</span> <span class="o">=</span> <span class="k">of</span><span class="p">(</span><span class="mi">1</span><span class="p">,</span> <span class="mi">2</span><span class="p">);</span>
<span class="kr">const</span> <span class="nx">obs$</span> <span class="o">=</span> <span class="nx">zip</span><span class="p">(</span><span class="nx">obs1</span><span class="p">,</span> <span class="nx">obs2</span><span class="p">).</span><span class="nx">pipe</span><span class="p">(</span>
  <span class="nx">map</span><span class="p">(</span><span class="nx">values</span> <span class="o">=&gt;</span> <span class="nx">values</span><span class="p">[</span><span class="mi">1</span><span class="p">])</span>
<span class="p">);</span>

<span class="kr">const</span> <span class="nx">published</span> <span class="o">=</span> <span class="nx">obs$</span>
  <span class="p">.</span><span class="nx">pipe</span><span class="p">(</span>
    <span class="nx">publish</span><span class="p">(</span>
      <span class="nx">$multicasted</span> <span class="o">=&gt;</span> <span class="nx">merge</span><span class="p">(</span>
        <span class="c1">// 多次使用$multicasted其数据来自同一数据流</span>
        <span class="nx">$multicasted</span><span class="p">.</span><span class="nx">pipe</span><span class="p">(</span><span class="nx">tap</span><span class="p">(</span><span class="nx">x</span> <span class="o">=&gt;</span> <span class="nx">console</span><span class="p">.</span><span class="nx">log</span><span class="p">(</span><span class="sb">`Tap x1: </span><span class="si">${</span><span class="nx">x</span><span class="si">}</span><span class="sb">`</span><span class="p">))),</span>
        <span class="nx">$multicasted</span><span class="p">.</span><span class="nx">pipe</span><span class="p">(</span>
          <span class="nx">map</span><span class="p">(</span><span class="nx">x</span> <span class="o">=&gt;</span> <span class="nx">x</span> <span class="o">*</span> <span class="mi">10</span><span class="p">),</span>
          <span class="nx">tap</span><span class="p">(</span><span class="nx">x</span> <span class="o">=&gt;</span> <span class="nx">console</span><span class="p">.</span><span class="nx">log</span><span class="p">(</span><span class="sb">`Tap x10: </span><span class="si">${</span><span class="nx">x</span><span class="si">}</span><span class="sb">`</span><span class="p">)),</span>
        <span class="p">),</span>
      <span class="p">)</span>
    <span class="p">)</span>
  <span class="p">)</span>

<span class="nx">setTimeout</span><span class="p">(()</span> <span class="o">=&gt;</span> <span class="p">{</span>
  <span class="nx">published</span><span class="p">.</span><span class="nx">subscribe</span><span class="p">((</span><span class="nx">x</span><span class="p">)</span> <span class="o">=&gt;</span> <span class="nx">console</span><span class="p">.</span><span class="nx">log</span><span class="p">(</span><span class="sb">`Sub 1: </span><span class="si">${</span><span class="nx">x</span><span class="si">}</span><span class="sb">`</span><span class="p">));</span>
<span class="p">},</span> <span class="mi">1100</span><span class="p">);</span>

<span class="c1">// 输出</span>
<span class="c1">//</span>
<span class="c1">// Tap x1: 1</span>
<span class="c1">// Sub 1: 1</span>
<span class="c1">// Tap x10: 10</span>
<span class="c1">// Sub 1: 10</span>
<span class="c1">// Tap x1: 2</span>
<span class="c1">// Sub 1: 2</span>
<span class="c1">// Tap x10: 20</span>
<span class="c1">// Sub 1: 20</span>
</code></pre></div>


<p>另外，在传入selector参数时，<code>publish</code>运算符返回selector的返回值，这不一定是<code>ConnectableObserver</code>对象，也不一定是“热”的Observer：</p>
<div class="highlight"><pre><span></span><code><span class="kr">import</span> <span class="p">{</span> <span class="k">of</span><span class="p">,</span> <span class="nx">zip</span><span class="p">,</span> <span class="nx">interval</span><span class="p">,</span> <span class="nx">merge</span><span class="p">,</span> <span class="nx">ConnectableObservable</span> <span class="p">}</span> <span class="nx">from</span> <span class="s2">&quot;rxjs&quot;</span><span class="p">;</span>
<span class="kr">import</span> <span class="p">{</span> <span class="nx">map</span><span class="p">,</span> <span class="nx">publish</span><span class="p">,</span> <span class="nx">tap</span> <span class="p">}</span> <span class="nx">from</span> <span class="s2">&quot;rxjs/operators&quot;</span><span class="p">;</span>

<span class="nx">console</span><span class="p">.</span><span class="nx">clear</span><span class="p">();</span>

<span class="kr">const</span> <span class="nx">obs1</span> <span class="o">=</span> <span class="nx">interval</span><span class="p">(</span><span class="mi">1000</span><span class="p">);</span>
<span class="kr">const</span> <span class="nx">obs2</span> <span class="o">=</span> <span class="k">of</span><span class="p">(</span><span class="mi">1</span><span class="p">,</span> <span class="mi">2</span><span class="p">);</span>
<span class="kr">const</span> <span class="nx">obs$</span> <span class="o">=</span> <span class="nx">zip</span><span class="p">(</span><span class="nx">obs1</span><span class="p">,</span> <span class="nx">obs2</span><span class="p">).</span><span class="nx">pipe</span><span class="p">(</span>
  <span class="nx">map</span><span class="p">(</span><span class="nx">values</span> <span class="o">=&gt;</span> <span class="nx">values</span><span class="p">[</span><span class="mi">1</span><span class="p">])</span>
<span class="p">);</span>

<span class="kr">const</span> <span class="nx">published</span> <span class="o">=</span> <span class="nx">obs$</span>
  <span class="p">.</span><span class="nx">pipe</span><span class="p">(</span>
    <span class="nx">publish</span><span class="p">(</span>
      <span class="nx">$multicasted</span> <span class="o">=&gt;</span> <span class="nx">$multicasted</span><span class="p">.</span><span class="nx">pipe</span><span class="p">(</span>
        <span class="nx">tap</span><span class="p">(</span><span class="nx">x</span> <span class="o">=&gt;</span> <span class="nx">console</span><span class="p">.</span><span class="nx">log</span><span class="p">(</span><span class="sb">`Tap : </span><span class="si">${</span><span class="nx">x</span><span class="si">}</span><span class="sb">`</span><span class="p">))</span>
      <span class="p">)</span>
    <span class="p">)</span>
  <span class="p">)</span>

<span class="c1">// 这两个subscription都独自收到了所有的数据，因此published是“冷”Observer</span>
<span class="nx">setTimeout</span><span class="p">(()</span> <span class="o">=&gt;</span> <span class="p">{</span>
  <span class="nx">published</span><span class="p">.</span><span class="nx">subscribe</span><span class="p">((</span><span class="nx">x</span><span class="p">)</span> <span class="o">=&gt;</span> <span class="nx">console</span><span class="p">.</span><span class="nx">log</span><span class="p">(</span><span class="sb">`Sub 1: </span><span class="si">${</span><span class="nx">x</span><span class="si">}</span><span class="sb">`</span><span class="p">));</span>
  <span class="nx">setTimeout</span><span class="p">(()</span> <span class="o">=&gt;</span> <span class="p">{</span>
    <span class="nx">published</span><span class="p">.</span><span class="nx">subscribe</span><span class="p">((</span><span class="nx">x</span><span class="p">)</span> <span class="o">=&gt;</span> <span class="nx">console</span><span class="p">.</span><span class="nx">log</span><span class="p">(</span><span class="sb">`Sub 2: </span><span class="si">${</span><span class="nx">x</span><span class="si">}</span><span class="sb">`</span><span class="p">));</span>
  <span class="p">},</span> <span class="mi">1100</span><span class="p">);</span>
<span class="p">},</span> <span class="mi">1100</span><span class="p">);</span>

<span class="k">if</span> <span class="p">(</span><span class="k">typeof</span> <span class="nx">published</span><span class="p">.</span><span class="nx">connect</span> <span class="o">===</span> <span class="s1">&#39;function&#39;</span><span class="p">)</span> <span class="p">{</span>
  <span class="nx">published</span><span class="p">.</span><span class="nx">connect</span><span class="p">();</span>
  <span class="nx">console</span><span class="p">.</span><span class="nx">log</span><span class="p">(</span><span class="s1">&#39;Connected&#39;</span><span class="p">);</span>
<span class="p">}</span> <span class="k">else</span> <span class="p">{</span>
  <span class="c1">// 这不是ConnectableObservable对象</span>
  <span class="nx">console</span><span class="p">.</span><span class="nx">log</span><span class="p">(</span><span class="s1">&#39;Not connectable&#39;</span><span class="p">);</span>
<span class="p">}</span>

<span class="c1">// 输出：</span>
<span class="c1">//</span>
<span class="c1">// Not connectable</span>
<span class="c1">// Tap : 1</span>
<span class="c1">// Sub 1: 1</span>
<span class="c1">// Tap : 2</span>
<span class="c1">// Sub 1: 2</span>
<span class="c1">// Tap : 1</span>
<span class="c1">// Sub 2: 1</span>
<span class="c1">// Tap : 2</span>
<span class="c1">// Sub 2: 2</span>
</code></pre></div>


<p><code>ConnectableObservable</code>对象可以和<code>refCount</code>运算符串联使用，<code>refCount</code>运算符返回一个Observable对象，该对象的subscription数量&gt;0时，<code>refCount</code>上游的<code>ConnectableObservable</code>对象连接到源，该对象的subscription数量减少到0时，其上游的<code>ConnectableObservable</code>对象从源断开连接。</p>
<div class="highlight"><pre><span></span><code><span class="kr">import</span> <span class="p">{</span> <span class="nx">interval</span> <span class="p">}</span> <span class="nx">from</span> <span class="s1">&#39;rxjs&#39;</span><span class="p">;</span>
<span class="kr">import</span> <span class="p">{</span> <span class="nx">tap</span><span class="p">,</span> <span class="nx">publish</span><span class="p">,</span> <span class="nx">refCount</span> <span class="p">}</span> <span class="nx">from</span> <span class="s1">&#39;rxjs/operators&#39;</span><span class="p">;</span>

<span class="kr">const</span> <span class="nx">obs$</span> <span class="o">=</span> <span class="nx">interval</span><span class="p">(</span><span class="mi">1000</span><span class="p">).</span><span class="nx">pipe</span><span class="p">(</span>
  <span class="nx">tap</span><span class="p">(</span><span class="nx">x</span> <span class="o">=&gt;</span> <span class="nx">console</span><span class="p">.</span><span class="nx">log</span><span class="p">(</span><span class="sb">`Tap: </span><span class="si">${</span><span class="nx">x</span><span class="si">}</span><span class="sb">`</span><span class="p">)),</span>
  <span class="nx">publish</span><span class="p">(),</span>
  <span class="nx">refCount</span><span class="p">(),</span>
<span class="p">)</span>

<span class="nx">setTimeout</span><span class="p">(()</span> <span class="o">=&gt;</span> <span class="p">{</span>
  <span class="c1">// 开始产生数据</span>
  <span class="kr">const</span> <span class="nx">sub</span> <span class="o">=</span> <span class="nx">obs$</span><span class="p">.</span><span class="nx">subscribe</span><span class="p">();</span>
  <span class="nx">setTimeout</span><span class="p">(()</span> <span class="o">=&gt;</span> <span class="p">{</span>
    <span class="c1">// 停止产生数据</span>
    <span class="nx">sub</span><span class="p">.</span><span class="nx">unsubscribe</span><span class="p">();</span>
  <span class="p">},</span> <span class="mi">2100</span><span class="p">);</span>
<span class="p">},</span> <span class="mi">1100</span><span class="p">);</span>

<span class="c1">// 输出</span>
<span class="c1">//</span>
<span class="c1">// Tap: 0</span>
<span class="c1">// Tap: 1</span>
</code></pre></div>


<p><code>refCount</code>返回的Observable对象的subscription数量减少到0之后，如果又有新的subscription，则前后两段subscription接收到的数据流是否相同取决于<code>refCount</code>上游的<code>ConnectableObservable</code>对象的源是“冷”还是“热”。</p>
<div class="highlight"><pre><span></span><code><span class="kr">import</span> <span class="p">{</span><span class="nx">interval</span><span class="p">}</span> <span class="nx">from</span> <span class="s1">&#39;rxjs&#39;</span><span class="p">;</span>
<span class="kr">import</span> <span class="p">{</span><span class="nx">take</span><span class="p">,</span> <span class="nx">publish</span><span class="p">,</span> <span class="nx">refCount</span><span class="p">}</span> <span class="nx">from</span> <span class="s1">&#39;rxjs/operators&#39;</span><span class="p">;</span>

<span class="kr">const</span> <span class="nx">coldSource</span> <span class="o">=</span> <span class="nx">interval</span><span class="p">(</span><span class="mi">1000</span><span class="p">).</span><span class="nx">pipe</span><span class="p">(</span>
  <span class="nx">take</span><span class="p">(</span><span class="mi">3</span><span class="p">)</span>
<span class="p">);</span>
<span class="kr">const</span> <span class="nx">hotSource</span> <span class="o">=</span> <span class="nx">coldSource</span><span class="p">.</span><span class="nx">pipe</span><span class="p">(</span>
  <span class="nx">publish</span><span class="p">()</span>
<span class="p">);</span>

<span class="kr">const</span> <span class="nx">obsWithColdSource</span> <span class="o">=</span> <span class="nx">coldSource</span><span class="p">.</span><span class="nx">pipe</span><span class="p">(</span>
  <span class="nx">publish</span><span class="p">(),</span>
  <span class="nx">refCount</span><span class="p">(),</span>
<span class="p">);</span>

<span class="kr">const</span> <span class="nx">obsWithHotSource</span> <span class="o">=</span> <span class="nx">hotSource</span><span class="p">.</span><span class="nx">pipe</span><span class="p">(</span>
  <span class="nx">publish</span><span class="p">(),</span>
  <span class="nx">refCount</span><span class="p">(),</span>
<span class="p">);</span>

<span class="kr">const</span> <span class="nx">cs1</span> <span class="o">=</span> <span class="nx">obsWithColdSource</span><span class="p">.</span><span class="nx">subscribe</span><span class="p">(</span>
  <span class="nx">x</span> <span class="o">=&gt;</span> <span class="nx">console</span><span class="p">.</span><span class="nx">log</span><span class="p">(</span><span class="sb">`Cold 1: </span><span class="si">${</span><span class="nx">x</span><span class="si">}</span><span class="sb">`</span><span class="p">)</span>
<span class="p">);</span>
<span class="kr">const</span> <span class="nx">hs1</span> <span class="o">=</span> <span class="nx">obsWithHotSource</span><span class="p">.</span><span class="nx">subscribe</span><span class="p">(</span>
  <span class="nx">x</span> <span class="o">=&gt;</span> <span class="nx">console</span><span class="p">.</span><span class="nx">log</span><span class="p">(</span><span class="sb">`Hot 1: </span><span class="si">${</span><span class="nx">x</span><span class="si">}</span><span class="sb">`</span><span class="p">)</span>
<span class="p">);</span>
<span class="nx">setTimeout</span><span class="p">(()</span> <span class="o">=&gt;</span> <span class="p">{</span>
  <span class="nx">cs1</span><span class="p">.</span><span class="nx">unsubscribe</span><span class="p">();</span>
  <span class="c1">// 输出新数据流中的内容</span>
  <span class="kr">const</span> <span class="nx">cs2</span> <span class="o">=</span> <span class="nx">obsWithColdSource</span><span class="p">.</span><span class="nx">subscribe</span><span class="p">(</span>
    <span class="nx">x</span> <span class="o">=&gt;</span> <span class="nx">console</span><span class="p">.</span><span class="nx">log</span><span class="p">(</span><span class="sb">`Cold 2: </span><span class="si">${</span><span class="nx">x</span><span class="si">}</span><span class="sb">`</span><span class="p">)</span>
  <span class="p">)</span>
  <span class="nx">hs1</span><span class="p">.</span><span class="nx">unsubscribe</span><span class="p">();</span>
  <span class="c1">// 继续输出原先数据流中的内容</span>
  <span class="kr">const</span> <span class="nx">hs2</span> <span class="o">=</span> <span class="nx">obsWithHotSource</span><span class="p">.</span><span class="nx">subscribe</span><span class="p">(</span>
    <span class="nx">x</span> <span class="o">=&gt;</span> <span class="nx">console</span><span class="p">.</span><span class="nx">log</span><span class="p">(</span><span class="sb">`Hot 2: </span><span class="si">${</span><span class="nx">x</span><span class="si">}</span><span class="sb">`</span><span class="p">)</span>
  <span class="p">);</span>
<span class="p">},</span> <span class="mi">1100</span><span class="p">);</span>

<span class="nx">hotSource</span><span class="p">.</span><span class="nx">connect</span><span class="p">();</span>

<span class="c1">// 输出：</span>
<span class="c1">//</span>
<span class="c1">// Cold 1: 0</span>
<span class="c1">// Hot 1: 0</span>
<span class="c1">// Hot 2: 1</span>
<span class="c1">// Cold 2: 0</span>
<span class="c1">// Hot 2: 2</span>
<span class="c1">// Cold 2: 1</span>
<span class="c1">// Cold 2: 2</span>
</code></pre></div>


<p><code>publish + refCount</code>运算符的组合很常见，以至于有专门的<code>share</code>运算符来表示这种组合：</p>
<div class="highlight"><pre><span></span><code><span class="nx">obs$</span><span class="p">.</span><span class="nx">pipe</span><span class="p">(</span><span class="nx">share</span><span class="p">())</span>
<span class="c1">// 等价于</span>
<span class="nx">obs$</span><span class="p">.</span><span class="nx">pipe</span><span class="p">(</span><span class="nx">publish</span><span class="p">(),</span> <span class="nx">refCount</span><span class="p">())</span>
</code></pre></div>


<p><code>publishReplay(1)</code>和<code>publishLast()</code>这两种方式返回的Observable对象都会缓存上游数据流中的最新一份数据，不管当前是否有subscription存在，该Observable对象的新subscription会马上接收到其缓存的数据。两者的区别在于<code>publishLast()</code>等到上游的Observable对象完成，才缓存其完成前的最新一份数据，其subscription收到其缓存的数据后会马上收到其完成的信号。<code>publishReplay(1)</code>不必等到上游的Observable对象完成，其subscription收到其缓存的数据后，可能收到新的数据，也可能收到其完成的信号，取决于其上游的Observable对象是否完成。</p>
<div class="highlight"><pre><span></span><code><span class="kr">import</span> <span class="p">{</span> <span class="nx">interval</span> <span class="p">}</span> <span class="nx">from</span> <span class="s1">&#39;rxjs&#39;</span><span class="p">;</span>
<span class="kr">import</span> <span class="p">{</span> <span class="nx">take</span><span class="p">,</span> <span class="nx">publishLast</span><span class="p">,</span> <span class="nx">publishReplay</span><span class="p">,</span> <span class="nx">refCount</span> <span class="p">}</span> <span class="nx">from</span> <span class="s1">&#39;rxjs/operators&#39;</span><span class="p">;</span>

<span class="kr">const</span> <span class="nx">source</span> <span class="o">=</span> <span class="nx">interval</span><span class="p">(</span><span class="mi">1000</span><span class="p">).</span><span class="nx">pipe</span><span class="p">(</span>
  <span class="nx">take</span><span class="p">(</span><span class="mi">3</span><span class="p">)</span>
<span class="p">);</span>

<span class="kr">const</span> <span class="nx">lastObs</span> <span class="o">=</span> <span class="nx">source</span><span class="p">.</span><span class="nx">pipe</span><span class="p">(</span>
  <span class="nx">publishLast</span><span class="p">(),</span>
  <span class="nx">refCount</span><span class="p">(),</span>
<span class="p">);</span>

<span class="kr">const</span> <span class="nx">replayObs</span> <span class="o">=</span> <span class="nx">source</span><span class="p">.</span><span class="nx">pipe</span><span class="p">(</span>
  <span class="nx">publishReplay</span><span class="p">(</span><span class="mi">1</span><span class="p">),</span>
  <span class="nx">refCount</span><span class="p">(),</span>
<span class="p">);</span>

<span class="nx">setTimeout</span><span class="p">(()</span> <span class="o">=&gt;</span> <span class="p">{</span>
  <span class="c1">// 只收到数据流中最后一个数据，然后马上complete</span>
  <span class="nx">lastObs</span><span class="p">.</span><span class="nx">subscribe</span><span class="p">({</span>
    <span class="nx">next</span>: <span class="kt">x</span> <span class="o">=&gt;</span> <span class="nx">console</span><span class="p">.</span><span class="nx">log</span><span class="p">(</span><span class="sb">`Last: </span><span class="si">${</span><span class="nx">x</span><span class="si">}</span><span class="sb">`</span><span class="p">),</span>
    <span class="nx">complete</span><span class="o">:</span> <span class="p">()</span> <span class="o">=&gt;</span> <span class="nx">console</span><span class="p">.</span><span class="nx">log</span><span class="p">(</span><span class="s1">&#39;Last complete&#39;</span><span class="p">)</span>
  <span class="p">});</span>
  <span class="c1">// 收到数据流中缓存的最新数据以及之后的数据，直至complete</span>
  <span class="nx">replayObs</span><span class="p">.</span><span class="nx">subscribe</span><span class="p">({</span>
    <span class="nx">next</span>: <span class="kt">x</span> <span class="o">=&gt;</span> <span class="nx">console</span><span class="p">.</span><span class="nx">log</span><span class="p">(</span><span class="sb">`Replay: </span><span class="si">${</span><span class="nx">x</span><span class="si">}</span><span class="sb">`</span><span class="p">),</span>
    <span class="nx">complete</span><span class="o">:</span> <span class="p">()</span> <span class="o">=&gt;</span> <span class="nx">console</span><span class="p">.</span><span class="nx">log</span><span class="p">(</span><span class="s1">&#39;Replay complete&#39;</span><span class="p">)</span>
  <span class="p">});</span>
<span class="p">},</span> <span class="mi">2100</span><span class="p">);</span>

<span class="c1">// 开始产生数据</span>
<span class="nx">lastObs</span><span class="p">.</span><span class="nx">subscribe</span><span class="p">();</span>
<span class="nx">replayObs</span><span class="p">.</span><span class="nx">subscribe</span><span class="p">();</span>

<span class="c1">// 输出：</span>
<span class="c1">//</span>
<span class="c1">// Replay: 1</span>
<span class="c1">// Last: 2</span>
<span class="c1">// Last complete</span>
<span class="c1">// Replay: 2</span>
<span class="c1">// Replay complete</span>
</code></pre></div>


<p><code>publishReplay</code>和<code>refCount</code>的组合也很常见，因此有单独的运算符<code>shareReplay</code>来表示二者的组合。</p>
<h4>tap运算符</h4>
<p><code>tap</code>运算符对上游数据流中的每一项数据调用指定的处理函数进行处理，然后将数据原封不动地发送给下游，处理函数的返回值（如果有的话）被丢弃，只为其副作用（如打印日志）。</p>
<p>上游的错误也会被原封不动地发送给下游，处理函数中发生的错误也会被发送到下游。无论是上游出现错误还是tap处理函数中发生了错误，数据流都会终止。</p>
<div class="highlight"><pre><span></span><code><span class="kr">import</span> <span class="p">{</span> <span class="k">of</span><span class="p">,</span> <span class="nx">throwError</span><span class="p">,</span> <span class="nx">concat</span> <span class="p">}</span> <span class="nx">from</span> <span class="s1">&#39;rxjs&#39;</span><span class="p">;</span>
<span class="kr">import</span> <span class="p">{</span> <span class="nx">tap</span> <span class="p">}</span> <span class="nx">from</span> <span class="s1">&#39;rxjs/operators&#39;</span><span class="p">;</span>

<span class="kr">const</span> <span class="nx">processStream</span> <span class="o">=</span> <span class="p">(</span><span class="nx">source</span><span class="p">)</span> <span class="o">=&gt;</span> <span class="p">{</span>
  <span class="nx">source</span><span class="p">.</span><span class="nx">pipe</span><span class="p">(</span>
    <span class="nx">tap</span><span class="p">(</span><span class="nx">x</span> <span class="o">=&gt;</span> <span class="p">{</span>
      <span class="k">if</span> <span class="p">(</span><span class="k">typeof</span> <span class="nx">x</span> <span class="o">===</span> <span class="s1">&#39;number&#39;</span><span class="p">)</span> <span class="p">{</span>
        <span class="nx">console</span><span class="p">.</span><span class="nx">log</span><span class="p">(</span><span class="sb">`Tap, number: </span><span class="si">${</span><span class="nx">x</span><span class="si">}</span><span class="sb">`</span><span class="p">);</span>
      <span class="p">}</span> <span class="k">else</span> <span class="p">{</span>
        <span class="c1">// 该错误会被发送到下游，数据流终止</span>
        <span class="k">throw</span> <span class="k">new</span> <span class="nb">Error</span><span class="p">(</span><span class="sb">`Tap: </span><span class="si">${</span><span class="nx">x</span><span class="si">}</span><span class="sb"> is not number`</span><span class="p">);</span>
      <span class="p">}</span>
    <span class="p">},</span> <span class="nx">e</span> <span class="o">=&gt;</span> <span class="p">{</span>
      <span class="c1">// 处理上游的错误，下游仍会收到同样的错误，数据流终止</span>
      <span class="nx">console</span><span class="p">.</span><span class="nx">log</span><span class="p">(</span><span class="sb">`Tap, handle error: </span><span class="si">${</span><span class="nx">e</span><span class="si">}</span><span class="sb">`</span><span class="p">)</span>
    <span class="p">})</span>
  <span class="p">).</span><span class="nx">subscribe</span><span class="p">({</span>
    <span class="nx">next</span>: <span class="kt">x</span> <span class="o">=&gt;</span> <span class="nx">console</span><span class="p">.</span><span class="nx">log</span><span class="p">(</span><span class="sb">`Sub, value: </span><span class="si">${</span><span class="nx">x</span><span class="si">}</span><span class="sb">`</span><span class="p">),</span>
    <span class="nx">error</span>: <span class="kt">e</span> <span class="o">=&gt;</span> <span class="nx">console</span><span class="p">.</span><span class="nx">log</span><span class="p">(</span><span class="sb">`Sub, error: </span><span class="si">${</span><span class="nx">e</span><span class="si">}</span><span class="sb">`</span><span class="p">),</span>
  <span class="p">});</span>
<span class="p">};</span>

<span class="nx">processStream</span><span class="p">(</span><span class="nx">concat</span><span class="p">(</span>
  <span class="k">of</span><span class="p">(</span><span class="mi">1</span><span class="p">,</span> <span class="mi">2</span><span class="p">),</span>
  <span class="nx">throwError</span><span class="p">(</span><span class="s1">&#39;Source: error&#39;</span><span class="p">),</span>
  <span class="k">of</span><span class="p">(</span><span class="mi">3</span><span class="p">,</span> <span class="s1">&#39;a&#39;</span><span class="p">),</span>
<span class="p">));</span>

<span class="nx">setTimeout</span><span class="p">(()</span> <span class="o">=&gt;</span> <span class="p">{</span>
  <span class="nx">processStream</span><span class="p">(</span><span class="nx">concat</span><span class="p">(</span>
    <span class="k">of</span><span class="p">(</span><span class="mi">1</span><span class="p">,</span> <span class="mi">2</span><span class="p">,</span> <span class="s1">&#39;a&#39;</span><span class="p">,</span> <span class="mi">3</span><span class="p">),</span>
    <span class="nx">throwError</span><span class="p">(</span><span class="s1">&#39;Source: error&#39;</span><span class="p">),</span>
  <span class="p">));</span>
<span class="p">},</span> <span class="mi">1000</span><span class="p">);</span>

<span class="c1">// 输出：</span>
<span class="c1">//</span>
<span class="c1">// Tap, number: 1</span>
<span class="c1">// Sub, value: 1</span>
<span class="c1">// Tap, number: 2</span>
<span class="c1">// Sub, value: 2</span>
<span class="c1">// Tap, handle error: Source: error</span>
<span class="c1">// Sub, error: Source: error</span>
<span class="c1">// Tap, number: 1</span>
<span class="c1">// Sub, value: 1</span>
<span class="c1">// Tap, number: 2</span>
<span class="c1">// Sub, value: 2</span>
<span class="c1">// Sub, error: Error: Tap: a is not number</span>
</code></pre></div>


<h4>Subject</h4>
<p>Subject是实现了<code>Observer</code>接口的<code>Observable</code>，因为RxJS中控制数据流的方式就是调用<code>Observer</code>接口中的方法，这就给了用户直接控制数据流的方式。同时这也意味着<code>Subject</code>一定是“热的”，因为其<code>Observer</code>接口方法被调用后，只有Subject当前的subscription及其下游可能看到数据的变动。之后的subscription即使看到以前的数据，也是<code>Subject</code>中缓存的数据，不是实时的数据。</p>
<p>RxJS 6提供了4种Subject：</p>
<ul>
<li><code>Subject</code>：下游能收到所有的实时数据，但不会收到缓存的数据</li>
<li><code>AsyncSubject</code>：下游能收到其完成前的最后一份数据</li>
<li><code>BehaviorSubject</code>：一定有一份初始数据，下游能收到<code>subscribe</code>调用时其最新的数据以及之后的数据</li>
<li><code>ReplaySubject</code>：下游能收到<code>subscribe</code>调用时，其缓存的指定数量的最新数据以及之后的数据</li>
</ul>
<p>它们之间的表现差异可以查看<a href="https://www.learnrxjs.io/subjects/">Learn RxJS的这篇文章</a>及<a href="https://stackblitz.com/edit/rxjs-subjects-comparison?file=index.ts&amp;devtoolsheight=100">实例演示</a>。</p>
<p><code>BehaviorSubject</code>和<code>ReplaySubject(1)</code>的区别在于前者可以确保有数据，后者不提供这种担保。因此<code>BehaviorSubject</code>提供了<code>value</code>属性，可以同步地获取其最新的数据，而其它类型的Subject只能异步地获取其数据。</p>
<p>Subject经常用来充当用户操作和数据流之间的桥梁，虽然<code>fromEvent</code>也能做到这一点，但使用Subject可以避免暴露控件底层的DOM元素细节，在大多数情况下是更好的选择。</p>
    </div>
<nav class="nav sidebar">
    <div class="siteinfo">
        <div class="sitename"><a href="../..">coinfaces</a></div>
        <div class="tagline">Another Side of Life</div>
    </div>

        <a href="../../categories/">categories</a>
        <a href="../..">&larr; Home</a>
</nav><footer class="footer">
    <span>
        &copy; jpli &middot;
        under <a rel="license" href="http://creativecommons.org/licenses/by-sa/4.0/">CC-BY-SA 4.0</a>
    </span>
    |
    <span>
        Powered by:
        <a href="http://blog.getpelican.com/">Pelican</a>
    </span>
</footer></body>
</html>